/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.changelog;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.changelog.pipe.PipeConnectionInfo;
import com.nhncorp.neptune.common.TimeLog;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.ipc.NRPC;
import com.nhncorp.neptune.tabletserver.ChangeLog;

public class ChangeLogClient {
  static final Log LOG = LogFactory.getLog(ChangeLogClient.class);

  static final byte[] MAGIC_KEY = new byte[] { 0xC, 0xA, 0xF, 0xE, 0xB, 0xA, 0xB, 0xE };
  
  private String tabletName;
  private SocketChannel sc;
  private InetSocketAddress[] pipeAddressList;
  private RpcUtil rpcUtil;
  private int seq = 0;
  private DataOutputStream currentStream;
  private InputStream checkInputStream;
  private int numWritten;
  private int timeout;
  private String[] ipAddressList;
  private boolean closed = true;
  private boolean startWriting = false;
  private String pipeKey;
  private static AtomicInteger pipeSeq = new AtomicInteger(0);
  
  private boolean consistencyCheck = false;

  private volatile long lastTouchedTime;

  List<ChangeLog> logList = new ArrayList<ChangeLog>();
  ByteArrayOutputStream bos = new ByteArrayOutputStream();

  volatile long rollbackPosition;
  
  // only for test
  public ChangeLogClient(NConfiguration conf, String tabletName) throws IOException {
    InetSocketAddress[] rpcAddressList = new InetSocketAddress[3];

    for (int i = 0; i < rpcAddressList.length; i++) {
      rpcAddressList[i] = new InetSocketAddress("127.0.0.1", 57001 + i);
    }

    init(conf, rpcAddressList, tabletName);
  }

  public ChangeLogClient(NConfiguration conf, String tabletName,
      InetSocketAddress[] addresses) throws IOException {
    LOG.debug("ChangeLogFileSystem instance is created");
    init(conf, addresses, tabletName);
  }

  public String getTabletName() {
    return tabletName;
  }

  private void init(NConfiguration conf, InetSocketAddress[] addressList, String tabletName)
      throws IOException {
    if (addressList == null) {
      throw new IOException("addresses is null");
    } else {
      for (int i = 0; i < addressList.length; i++) {
        if (addressList[i] == null) {
          throw new IOException("addresses contain null");
        }
      }
    }

    this.timeout = conf.getInt("neptune.changelog.server.pipe.timeout", 30) * 1000; // default 60 sec
    this.tabletName = tabletName;

    // 3 is num of replica
    // this.es = Executors.newFixedThreadPool(3);

    this.pipeAddressList = new InetSocketAddress[addressList.length];
    this.ipAddressList = new String[addressList.length];
    this.rpcUtil = new RpcUtil(conf, addressList);

    for (int i = 0; i < addressList.length; i++) {
      pipeAddressList[i] = new InetSocketAddress(addressList[i].getAddress(), (addressList[i]
          .getPort() + ChangeLogServer.PORT_DIFF));
      ipAddressList[i] = addressList[i].getAddress().getHostAddress() + ":"
          + (addressList[i].getPort() + ChangeLogServer.PORT_DIFF);
    }

    lastTouchedTime = System.currentTimeMillis();
  }
  
  public String getPipeKey() {
    return pipeKey;
  }

  public void open() throws UnmatchedLogException, ChangeLogInterruptedException, IOException {
    pipeKey = generateUniqueKey();

    if (LOG.isDebugEnabled()) {
      String msg = "";
      for (String addr : ipAddressList) {
        msg = msg + addr + ", ";
      }
      LOG.debug("Open new pipe with timeout[" + this.timeout + "], key [" + pipeKey + "], -> " + msg);
    }
    
    String ret = null;

    try {
      sc = SocketChannel.open();
      LOG.debug("open pipe to " + pipeAddressList[0]);
      sc.connect(pipeAddressList[0]);
      Socket s = sc.socket();
      s.setTcpNoDelay(true);
      s.setSoTimeout(timeout);
      s.setSendBufferSize(21900);
      s.setSendBufferSize(43800);

      byte[] body = buildBody();

      currentStream = new DataOutputStream(new BufferedOutputStream(s.getOutputStream(), 8192));
      checkInputStream = s.getInputStream();
      currentStream.write(MAGIC_KEY);
      currentStream.writeInt(0);
      currentStream.writeInt(body.length);
      currentStream.write(body);
      currentStream.flush();

      DataInputStream dis = new DataInputStream(s.getInputStream());
      byte[] key = new byte[MAGIC_KEY.length];
      dis.read(key);

      if (!Arrays.equals(MAGIC_KEY, key)) {
        throw new IOException("Fail to establish pipe connection to " + getPipeAddressList() + "due to wrong magic key");
      }
      int opCode = dis.readInt();
      int replyLength = dis.readInt();

      body = new byte[replyLength];
      dis.read(body);
      ret = new String(body);
    } catch (ClosedByInterruptException e) {
      internalClose();
      throw new ChangeLogInterruptedException();
    } catch (IOException e) {
      LOG.warn("Fail to establish pipe to " + getPipeAddressList() + " due to " + e);
      internalClose();
      throw new IOException("Fail to establish pipe connection to " + getPipeAddressList(), e);
    }
    
    if (!ret.startsWith("OK")) {
      internalClose();
      
      if (ret.equals("Log files are unmatched among change log servers")) {
        throw new UnmatchedLogException("Log files are unmatched among change log servers");
      }

      throw new IOException("Fail to establish pipe connection to " + getPipeAddressList()
          + ". response : " + ret);
    } else {
      closed = false;
      rollbackPosition = Long.parseLong(ret.substring(2));
    }
  }
  
  private String getPipeAddressList() {
    String msg = "";
    for (InetSocketAddress addr : pipeAddressList) {
      msg = msg + addr.getHostName() + ":" + addr.getPort() + ", ";
    }
    
    return msg;
  }

  public void startWriting(String id) throws ChangeLogInterruptedException, IOException {
    if (closed) {
      throw new IOException("Pipe is not yet established");
    }

    // 만일 이미 닫혀있는 Pipe라면 여기서 read를 통해 체크를 해줘야 한다.
    /*
     * while (checkInputStream.available() > 0) { try { LOG.debug("something in
     * the check input stream. it should be flushed"); int r =
     * checkInputStream.read(); } catch(IOException e) { LOG.info("reopen pipe
     * for id : " + id); internalClose(); open(); break; } }
     */

    if (sc.socket().isConnected() == false) {
      LOG.info("reopen pipe for id : " + id);
      internalClose();
      open();
    }

    byte[] idBytes = id.getBytes();

    try {
      currentStream.writeInt(seq++); // seq number
      currentStream.writeInt(idBytes.length);
      currentStream.write(idBytes); // id

      numWritten += (4 + 4 + idBytes.length);
      startWriting = true;
    } catch(ClosedByInterruptException e) {
      throw new ChangeLogInterruptedException();
    }
  }

  public void append(ChangeLog changeLog) {
    logList.add(changeLog);
    numWritten += changeLog.getByteSize();
  }

  public void write(byte[] bytes) {
    write(bytes, 0, bytes.length);
  }

  public void write(byte[] bytes, int offset, int length) {
    bos.write(bytes, offset, length);
    numWritten += length;
  }

  public int commitWriting() throws ChangeLogInterruptedException, IOException {
    int retCommit = 0;
    int ret = 0;
    try {
      numWritten += 4; // total length of change logs
      ret = numWritten;
      currentStream.writeInt(numWritten);

      // UnitTest를 위한 임시방편으로 이렇게 구현한다.
      for (ChangeLog log : logList) {
        log.write(currentStream);
      }

      if (bos.size() > 0) {
        currentStream.write(bos.toByteArray());
      }

      currentStream.flush();

      long startTime = System.currentTimeMillis();
      retCommit = checkInputStream.read();
      TimeLog.checkTime(startTime, 100, "checkInputStream.read()");
    } catch(ClosedByInterruptException e) {
      throw new ChangeLogInterruptedException();
    } catch (IOException e) {
      // try { Thread.sleep(1000); } catch(InterruptedException ex) { }
      LOG.warn("Exception in commiting changelog pipeKey[" + pipeKey + "] exception : " + e);
      
      try {
        rollback();
      } catch(ClosedByInterruptException ex) {
        throw new ChangeLogInterruptedException();
      } catch(Exception ex) {
        LOG.warn("Fail to rollback change log for tablet[" + tabletName + "] exception : " + e);
      }
      throw e;
    } finally {
      numWritten = 0;
      logList.clear();
      bos.reset();

      startWriting = false;
    }

    if ((retCommit & (byte) 0xff) != 0xff) {
      try {
        rollback();
      } catch(ClosedByInterruptException ex) {
        throw new ChangeLogInterruptedException();
      } catch(IOException e) {
        LOG.warn("Fail to rollback change log for tablet[" + tabletName + "]", e);
      }
      internalClose();
      LOG.warn("Fail to commit. commit ret : [" + retCommit 
          + "], change log rollback and pipe [" + pipeKey + "] is closed");
      throw new IOException("Commiting changelog is fail.");
    }

    rollbackPosition += (4 + ret);
    
    // ChangeLogFileChannel에서 실제 파일에 기록 할 때 
    // 현재 commit하는 txID에 해당하는 데이터의 총 바이트의 길이를 미리 데이터 앞에 기록한다.
    // 따라서 전체 기록양을 반환할 때는 반드시 4만큼 더해서 반환하도록 한다.
    return 4 + ret;
  }

  public boolean tabletExists() throws IOException {
    Method rpcMethod = rpcUtil.buildRpcMethod("tabletExists", String.class);
    
    for(int i = 0; i < rpcUtil.getMultiRpcCount(); i++) {
      Object ret = null;
      try {
        ret = rpcUtil.singleCall(rpcMethod, i, tabletName);
      } catch(IOException e) {
        LOG.warn("fail to check tablet [" + tabletName + "] exists in server[" + rpcUtil.getAddressAt(i) + "]due to " + e + ", but try next server");
        continue;
      }
      
      return ret.equals("true");
    }
    
    throw new IOException("Fail to check tablet exists to all servers");
  }
  
  public void prepareClosing() throws IOException {
    try {
      if (isPipeConnected()) {
        LOG.debug("Close pipe. key : " + pipeKey);
        if (isPipeConnected()) {
          Method rpcMethod = rpcUtil.buildRpcMethod("closePipe", String.class, String.class);
          rpcUtil.multiSerialCall(rpcMethod, tabletName, pipeKey);
        }
      } else {
        LOG.debug("Close pipe without calling rpc. key : " + pipeKey);
      }
    } catch (IOException e) {
      LOG.debug("Exception in closing pipe, it will be re-thrown, key : " + this.pipeKey, e);
      throw e;
    }
  }

  public void disconnect() throws IOException {
    internalClose();
  }
  
  public void close() throws IOException {
    prepareClosing();
    try {
      Thread.sleep(1000);
    } catch(InterruptedException e) {
      throw new ChangeLogInterruptedException();
    } finally {
      disconnect();
    }
  }

  private boolean isPipeConnected() {
    if (sc == null) {
      return false;
    }

    try {
      sc.configureBlocking(false);
    } catch (IOException e) {
      return false;
    }

    try {
      ByteBuffer buf = ByteBuffer.allocate(1);

      return sc.read(buf) >= 0;
    } catch (IOException e) {
      return false;
    }
  }

  private void internalClose() throws IOException {
    try {
      if (currentStream != null) {
        currentStream.close();
      }
  
      if (sc != null) {
        try {
          sc.close();
        } catch (ClosedByInterruptException e) {
          throw new ChangeLogInterruptedException();
        } finally {
          sc = null;
        }
      }
    } finally {
      closed = true;
      LOG.debug("Pipe for tablet : " + tabletName + " is closed");
    }
  }

  public void removeAllLogs() throws IOException {
    Method rpcMethod = rpcUtil.buildRpcMethod("removeAllLogs", String.class);
    rpcUtil.multiParallelCall_v2(rpcMethod, tabletName);
  }

  public void removeBackupLogFiles() throws IOException {
    Method rpcMethod = rpcUtil.buildRpcMethod("removeBackupLogs", String.class);
    rpcUtil.multiParallelCall_v2(rpcMethod, tabletName);
  }

  public void rollback() throws IOException {
    Method rpcMethod = rpcUtil.buildRpcMethod("rollback", String.class, String.class);
    rpcUtil.multiSerialCall(rpcMethod, tabletName, String.valueOf(rollbackPosition));
  }

  public void backupLogFile() throws IOException {
    Method backupMethod = rpcUtil.buildRpcMethod("backupLogFile", String.class);
    try {
      rpcUtil.multiParallelCall_v2(backupMethod, tabletName);
    } catch (IOException e) {
      LOG.warn("Error in backing up log file", e);

      Method restoreMethod = rpcUtil.buildRpcMethod("restoreLogFile", String.class);

      try {
        rpcUtil.multiParallelCall_v2(restoreMethod, tabletName);
      } catch (IOException ex) {
        LOG.warn("Error in restoring log file", ex);
      }

      throw new IOException("Fail to backup log files in tablet : " + tabletName);
    }
  }

  public TransactionData[] readLastLogFile() throws IOException {
    return readData(rpcUtil.buildRpcMethod("readLastLog", String.class));
  }

  public TransactionData[] readAllLogFiles() throws IOException {
    return readData(rpcUtil.buildRpcMethod("readAllLogs", String.class));
  }

  private TransactionData[] readData(Method readMethod) throws IOException {
    return readDataFrom(readMethod, getReplicaIndexForReading());
  }

  private TransactionData[] readDataFrom(Method readMethod, int index) throws IOException {
    int port = -1;
    int replicaCount = rpcUtil.getMultiRpcCount();

    for (int count = 0; count < replicaCount; count++) {
      try {
        port = (Integer) rpcUtil.singleCall(readMethod, index, tabletName);
        if (port > 0) {
          break;
        }
      } catch (IOException e) {
        LOG.info("Exception in reading change log data : " + e);
      }
      
      if (++index % replicaCount == 0) {
        index = 0;
      }
    }

    if (port < 0) {
      return null;
    }

    SocketChannel socketChannel = SocketChannel.open();
    List<TransactionData> txDataList = null;
    try {
      LOG.debug("Read change log from " + rpcUtil.getAddressAt(index) + ":" + port);
      socketChannel.connect(new InetSocketAddress(rpcUtil.getAddressAt(index), port));
      DataInputStream dis = new DataInputStream(new BufferedInputStream(socketChannel.socket().getInputStream(), 8192));

      txDataList = readTxDataFrom(dis);
    } finally {
      socketChannel.socket().close();
      socketChannel.close();
    }

    return txDataList.toArray(new TransactionData[0]);
  }

  static List<TransactionData> readTxDataFrom(DataInputStream dis) throws IOException {
    List<TransactionData> txDataList = new ArrayList<TransactionData>();
    while (true) {
      int totalByteLength = 0;

      try {
        totalByteLength = dis.readInt();
      } catch (EOFException e) {
        break;
      }

      if (totalByteLength <= 0) {
        LOG.warn("totalByteLength is smaller than zero[" + totalByteLength + "]!!, ignore this data");
        break;
      }

      byte[] bytes = new byte[totalByteLength];
      dis.readFully(bytes);
      txDataList.add(TransactionData.createFrom(bytes));
    }
    
    return txDataList;
  }

  private int getReplicaIndexForReading() throws IOException {
    Method rpcMethod = rpcUtil.buildRpcMethod("getLogFileSize", String.class);

    long minimumFileSize = Long.MAX_VALUE;
    int replicaIndex = 0;

    for (int i = 0; i < rpcUtil.getMultiRpcCount(); i++) {
      long ret = 0;

      try {
        ret = (Long) rpcUtil.singleCall(rpcMethod, i, tabletName);
      } catch (IOException e) {
        continue;
      }

      if (ret < minimumFileSize) {
        replicaIndex = i;
      }
    }

    return replicaIndex;
  }
  
  public long[] getAllLogFileSize() throws IOException {
    Method rpcMethod = rpcUtil.buildRpcMethod("getLogFileSize", String.class);
    long[] ret = new long[rpcUtil.getMultiRpcCount()];

    for (int i = 0; i < rpcUtil.getMultiRpcCount(); i++) {
      try {
        ret[i] = (Long) rpcUtil.singleCall(rpcMethod, i, tabletName);
      } catch(Exception e) {
        LOG.warn("Fail to get the size of log file of tablet[" + tabletName 
            + "], addr[" + pipeAddressList[i] + "]");
        ret[i] = -1;
      }
    }
    
    return ret;
  }
  
  public long getLogFileSize() throws IOException {
    long[] sizes = getAllLogFileSize();
    long ret = -1;
    
    for (int i = 0; i < rpcUtil.getMultiRpcCount(); i++) {
      if (sizes[i] < 0) {
        continue;
      }
      
      if (ret >= 0 && sizes[i] != ret) {
        LOG.warn("Unmatched logs! File sizes are different. size1 : " + sizes[i] + ", size2 : "
            + ret);
      }
      
      if (ret < sizes[i]) {
        ret = sizes[i];
      }
    }

    if (ret < 0) {
      throw new IOException("Fail to get the size of log file of tablet [" + tabletName + "]");
    }
    
    return ret;
  }

  private byte[] buildBody() throws IOException {
    return PipeConnectionInfo.buildFrom(consistencyCheck, pipeKey, tabletName, ipAddressList);
  }

  private String generateUniqueKey() {
    return tabletName + "_" + pipeSeq.getAndIncrement() + "@" + getLocalHostIntAddress();
  }

  // for test methods

  public void test_rollbackFileAtIndex(int index) {
    try {
      Method m = rpcUtil.buildRpcMethod("rollback", String.class, String.class);
      rpcUtil.singleCall(m, index, tabletName, String.valueOf(rollbackPosition));
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  public boolean isClosed() {
    return closed;
  }

  public boolean isWritingStarted() {
    return startWriting;
  }

  // by sangchul
  // 멀티 쓰레드가 접근할 수 있다.
  public boolean isExpired() {
    return System.currentTimeMillis() - lastTouchedTime >= 10000; // 10 sec
  }

  // 멀티 쓰레드가 접근할 수 있다.
  public ChangeLogClient touch() {
    lastTouchedTime = System.currentTimeMillis();
    return this;
  }

  public static void formatLogsAt(NConfiguration conf, InetSocketAddress addr)
      throws IOException {
    Method method;
    try {
      method = ChangeLogServerIF.class.getMethod("format");
      Object target = NRPC.getProxyWithoutVersionChecking(ChangeLogServerIF.class,
          ChangeLogServerIF.versionID, addr, conf);

      method.invoke(target);
    } catch (Exception e) {
      LOG.error("Error in formatting log servers", e);
      throw new IOException(e);
    }
  }

  public void setConsistencyCheck(boolean flag) {
    this.consistencyCheck = flag;
  }
  
  public boolean getConsistencyCheck() {
    return this.consistencyCheck;
  }

  public boolean hasInvalidChangeLog() throws IOException {
    Method m = rpcUtil.buildRpcMethod("getLogFileSize", String.class);
    Object[] retList = rpcUtil.multiSerialCall(m, tabletName);
    
    return !(retList[0].equals(retList[1]) && retList[1].equals(retList[2]));
  }
  
  static int localHostIntAddress = -1;
  
  private static int getLocalHostIntAddress() {
    if (localHostIntAddress == -1) {
      byte[] address;
      try {
        address = InetAddress.getLocalHost().getAddress();
      } catch (UnknownHostException e) {
        LOG.warn("Fail to get local host address. Set 127.0.0.1");
        localHostIntAddress = 0x7F000001;
        return localHostIntAddress;
      }
      
      int ret = 0;
      for(int i = address.length - 1; i >= 0; i--) {
          ret = (ret << 8) + (address[i] & 0xff);
      }
    
      localHostIntAddress = ret;
    }
    
    return localHostIntAddress;
  }
}
